package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.model.common.constants.HotArticleConstants;
import com.heima.model.common.mess.ArticleVisitStreamMess;
import com.heima.model.common.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.springframework.stereotype.Component;

import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String,String>> {


    /**
     * 接收待处理的数据 from
     * @return
     */
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    /**
     * 发送处理成功之后的数据 topic to
     * @return
     */
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    /**
     * 实时流式处理数据
     * @param stream
     * @return
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        return stream.flatMapValues(new ValueMapper<String, Iterable<? extends String>>() {
            @Override
            public Iterable<? extends String> apply(String value) {
                if (StringUtils.isNotBlank(value)) {
                    UpdateArticleMess updateArticleMess = JSON.parseObject(value, UpdateArticleMess.class);

                    log.info("UpdateArticleMess id:{}, type:{}",updateArticleMess.getArticleId(), updateArticleMess.getType().name() );

                    // 拼装分组类型 id:type  213123123:VIEWS  213123123:LIKES
                    return Arrays.asList( updateArticleMess.getArticleId()+":"+updateArticleMess.getType().name() );
                }
                return Arrays.asList();
            }
        })
                .groupBy((key,value) ->{
                    // value: 213123123:VIEWS  难点
                    return value;
                })
                .windowedBy(TimeWindows.of(5000))
                .count()
                .toStream()
                .map((key,value)->{
                    //key-->mess.getArticleId()+":"+mess.getType()
                    //value  long类型  count之后的数值
                    log.info("处理后结果key:{}-----value:{}",key.key().toString(),value.toString());
                    return new KeyValue<String,String>(key.key().toString(), formatObj(key.key().toString(), value.toString()));
                });
    }

    /**
     *
     * @param key  mess.getArticleId()+":"+mess.getType()  213123123:VIEWS
     * @param value 数值
     * @return
     */
    private String formatObj(String key, String value) {
        // 点赞多少次
        // 阅读多少次
        ArticleVisitStreamMess visitStreamMess = new ArticleVisitStreamMess();
        String[] split = key.split(":");  // 0:id 1:type
        visitStreamMess.setArticleId(Long.valueOf(split[0]));

        if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(split[1])) {
            visitStreamMess.setView(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(split[1])) {
            visitStreamMess.setLike(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(split[1])) {
            visitStreamMess.setComment(Long.valueOf(value));
        }

        if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(split[1])) {
            visitStreamMess.setCollect(Long.valueOf(value));
        }

        log.info("formatObj aggs score ArticleVisitStreamMess:{}", visitStreamMess);
        return JSON.toJSONString(visitStreamMess);
    }
}
